package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSQLApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderPaySucDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTradeOrderPaySucDetail().start(10016,4, Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv, String groupId) {

        // 1. 读取下单事务事实表
        //创建topic框架 读取
        sourceOrder_detail(tEnv);

        // 2. 读取 topic_db
        readOdsDb(tEnv,groupId);

        // 3. 读取 字典表
        createBaseDic(tEnv);

        // 4. 从 topic_db 中过滤 payment_info
        paymentStream(tEnv);

        // 5. 3张join: interval join 无需设置 ttl

        Table table = JOINpayment_oder_detail(tEnv);

        tEnv.createTemporaryView("pay_order",table);

        // 6、关联后数据做Look Join
        Table lookjoin = lookjoin(tEnv);

        // 6. 写出到 kafka 中

        //创建 topic框架
        createpayment_succ(tEnv);

        //插入
        lookjoin.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS).execute();


    }

    private static void createpayment_succ(StreamTableEnvironment tEnv) {
        TableResult tableResult = tEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS + "\n" +
                "(\n" +
                " id string,\n" +
                " order_id string,\n" +
                " sku_id string,\n" +
                " user_id string,\n" +
                " province_id string,\n" +
                " activity_id string,\n" +
                " activity_rule_id string,\n" +
                " coupon_id string,\n" +
                " sku_name string,\n" +
                " order_price string,\n" +
                " sku_num string,\n" +
                " split_total_amount string,\n" +
                " split_activity_amount string,\n" +
                " split_coupon_amount string,\n" +
                " payment_type_code string,\n" +
                " payment_type_name string,\n" +
                " payment_time string,\n" +
                " ts bigint\n" +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS));
    }

    private static Table lookjoin(StreamTableEnvironment tEnv) {
        return tEnv.sqlQuery("select\n" +
                "id,\n" +
                "order_id,\n" +
                "sku_id,\n" +
                "user_id,\n" +
                "province_id,\n" +
                "activity_id,\n" +
                "activity_rule_id,\n" +
                "coupon_id,\n" +
                "sku_name,\n" +
                "order_price,\n" +
                "sku_num,\n" +
                "split_total_amount,\n" +
                "split_activity_amount,\n" +
                "split_coupon_amount,\n" +
                "payment_type payment_type_code,\n" +
                "info.dic_name payment_type_name,\n" +
                "payment_time,\n" +
                "ts\n" +
                "from pay_order as p\n" +
                "join base_dic for system_time as of p.proc_time as b\n" +
                "on p.payment_type=b.rowkey");
    }

    private static Table JOINpayment_oder_detail(StreamTableEnvironment tEnv) {
        return tEnv.sqlQuery("select \n" +
                "o.id,\n" +
                "o.order_id,\n" +
                "sku_id,\n" +
                "p.user_id,\n" +
                "province_id,\n" +
                "activity_id,\n" +
                "activity_rule_id,\n" +
                "coupon_id,\n" +
                "sku_name,\n" +
                "order_price,\n" +
                "sku_num,\n" +
                "split_total_amount,\n" +
                "split_activity_amount,\n" +
                "split_coupon_amount,\n" +
                "p.ts,\n" +
                "payment_type,\n" +
                "callback_time payment_time,\n" +
                "proc_time\n" +
                "FROM payment_info p,dwd_trade_order_detail o\n" +
                "WHERE p.order_id = o.order_id\n" +
                "AND p.row_time BETWEEN o.row_time - INTERVAL '15' MINUTE AND p.row_time + INTERVAL '5' SECOND\n");
    }

    private static void paymentStream(StreamTableEnvironment tEnv) {
        Table table = tEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['payment_type'] payment_type,\n" +
                "`data`['callback_time'] callback_time,\n" +
                " ts,\n" +
                " row_time,\n" +
                " proc_time\n" +
                "from topic_db\n" +
                "where `database`='gmall' and `table` ='payment_info'\n" +
                "and `type` ='update'\n" +
                "and `old`['payment_status'] is not null\n" +
                "and `data`['payment_status'] ='1602'");

        tEnv.createTemporaryView("payment_info",table);
    }

    private static void sourceOrder_detail(StreamTableEnvironment tEnv) {
        TableResult tableResult = tEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_ORDER_DETAIL + "\n" +
                "(\n" +
                " id string,\n" +
                " order_id string,\n" +
                " sku_id string,\n" +
                " user_id string,\n" +
                " province_id string,\n" +
                " activity_id string,\n" +
                " activity_rule_id string,\n" +
                " coupon_id string,\n" +
                " sku_name string,\n" +
                " order_price string,\n" +
                " sku_num string,\n" +
                " create_time string,\n" +
                " split_total_amount string,\n" +
                " split_activity_amount string,\n" +
                " split_coupon_amount string,\n" +
                " split_original_amount string,\n" +
                " ts bigint,\n" +
                " row_time as TO_TIMESTAMP_LTZ(ts * 1000,3) ,\n"+
                " WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND\n"+
                ")" + SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS));
    }
}
